package com.atguigu.hotitems_analysis

import java.util.Properties

import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord}

/**
 * Kafka Producer生产者
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.hotitems_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/10 20:44
 */
object KafkaUtils {

  def KafkaSendUtils(): Unit = {
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop105:9092")
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    //创建kafka连接器
    val value = new KafkaProducer[String, String](properties)

    val path: String = KafkaUtils.getClass.getClassLoader.getResource("UserBehavior.csv").getPath

    //读取文件所有数据
    val line: Iterator[String] = io.Source.fromFile(path).getLines()

    for (elem <- line) {
      val lines = new ProducerRecord[String, String]("hotitmes", elem)
      //发送数据到kafka
      value.send(lines)
    }


    value.close()
  }


  def main(args: Array[String]): Unit = {
    KafkaSendUtils()
  }
}
